/*
 * This file is part of b.AMP.
 *
 * b.AMP is free software: you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * b.AMP is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with b.AMP.  If not, see <http://www.gnu.org/licenses/>.
 */

#include <bamp/basic_message.hpp>

#include <service/process.hpp>
#include <service/message_target.hpp>

#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
#include <boost/test/unit_test.hpp>

#include <iostream>
#include <string>
#include <queue>


/**
 * Naive serializer function that writes a message's contents on a
 * single line. Fields are delimited with a tabulation. The message
 * must not contain spaces or tabs!
 */
void serialize(std::ostream& stream, const bamp::Message& msg)
{
    stream << "begin msg\t" << msg.getName() << "\t";
    for (size_t i = 0; i < msg.getArgc(); ++i)
    {
        stream << "\t" << msg.getArgName(i)
               << "\t" << msg.getArg(i).typeName()
               << "\t" << msg.getArg(i).string();
    }
    stream << "\tend of msg" << std::endl;
}

/**
 * Deserializer that supports the same format as serialize().
 */
void deserialize(std::istream& stream, bamp::MessageTarget& target)
{
    std::string tok[3];
    bamp::BasicMessage msg;
    while ( (stream >> tok[0]) && (stream >> tok[1]) && (stream >> tok[2]) )
    {
        if (tok[0] == "begin" && tok[1] == "msg")
        {
            msg = bamp::BasicMessage(tok[2]);
        }
        else if (tok[0] == "end" && tok[1] == "of" && tok[2] == "msg")
        {
            target.send(msg);
        }
        else
        {
            bamp::Value value(tok[1], tok[2]);
            msg.setArg(tok[0], value);
        }
    }
}


/**
 * Class that receives messages and stores them until someone gets
 * them.
 */
class MessageQueue : public bamp::MessageTarget
{
public:

    /**
     * Wait until a message is available or until timeout was reached.
     * @param timeout maximum time to wait, in milliseconds
     * @return        true if a message was received before timeout
     */
    bool wait(int timeout)
    {
        const boost::system_time endtime
            = boost::get_system_time()
            + boost::posix_time::milliseconds(timeout);

        boost::mutex::scoped_lock lock(mutex);
        while (queue.size() == 0)
        {
            if (!notify.timed_wait(lock, endtime)) return false;
        }
        return true;
    }

    boost::shared_ptr<bamp::Message> pop()
    {
        boost::shared_ptr<bamp::Message> msg = queue.front();
        queue.pop();
        return msg;
    }
    

    /**
     * @return number of messages queued
     */
    bool size()
    {
        boost::mutex::scoped_lock lock(mutex);
        return queue.size();
    }

    virtual void send(const bamp::Message& message, bamp::MessageAck ack)
    {
        boost::mutex::scoped_lock lock(mutex);
        queue.push(boost::shared_ptr<bamp::Message>(message.clone()));
        ack.ack();
        notify.notify_one();
    }

    virtual bamp::MessageAck send(const bamp::Message& message)
    {
        bamp::MessageAck ack(1);
        send(message, ack);
        return ack;
    }


private:

    typedef std::queue<boost::shared_ptr<bamp::Message> > Queue;
    Queue                     queue;
    boost::mutex              mutex;
    boost::condition_variable notify;
};


BOOST_AUTO_TEST_CASE( process )
{
    bamp::Process::Descriptor desc = {"./echo"};
    bamp::Process proc(desc, serialize, deserialize);
    MessageQueue queue;
    
    proc.setTarget(queue);
    proc.run();

    BOOST_CHECK_EQUAL(queue.size(), 0);
    bamp::BasicMessage msg1("basic_message_1");
    msg1.setArg("arg_1", "value_1");
    msg1.setArg("arg_2", "value_2");
    proc.send(msg1);

    std::cout << "Sending:  ";
    serialize(std::cout, msg1);

    bool hasMsg = queue.wait(500);
    BOOST_CHECK_EQUAL(hasMsg, true);

    if (hasMsg)
    {
        boost::shared_ptr<bamp::Message> msg2 = queue.pop();

        std::cout << "Received: ";
        serialize(std::cout, *msg2);

        BOOST_CHECK_EQUAL(msg1.getName(),     msg2->getName());
        BOOST_CHECK_EQUAL(msg1.getArgName(0), msg2->getArgName(0));
        BOOST_CHECK_EQUAL(msg1.getArg(0),     msg2->getArg(0));
    }
}
